package com.beta.cms.test.demo.kafka.service.impl;

import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

//@Service
public class ConsumerKafkaMessageListener implements InitializingBean, DisposableBean {
	private static final Logger logger = LoggerFactory.getLogger(ConsumerKafkaMessageListener.class);
	KafkaConsumer<String, String> consumer;

	private void createConsumer() {
		Properties properties = new Properties();
		// Point it to the brokers
		properties.setProperty("bootstrap.servers", "localhost:9092");
		// Set the consumer group (all consumers must belong to a group).
		properties.setProperty("group.id", "groupA");
		// Set how to serialize key/value pairs
		properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		// When a group is first created, it has no offset stored to start
		// reading from. This tells it to start
		// with the earliest record in the stream.
		properties.setProperty("auto.offset.reset", "earliest");
		consumer = new KafkaConsumer<>(properties);
		consumer.subscribe(Arrays.asList("my-replicated-topic"));
	}

	private void startService() {
		// logger.info("consumer service started");
		while (true) {
			// Poll for records
			ConsumerRecords<String, String> records = consumer.poll(200);
			for (ConsumerRecord<String, String> record : records) {
				// logger.info(record.value());
				System.out.println(record.value() + "========================================");
			}
		}
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		createConsumer();
		startService();
	}

	@Override
	public void destroy() throws Exception {
		if (consumer != null) {
			consumer.close();
			consumer = null;
		}
	}
}
